package chenxu.test.hbase.storm;

import java.io.IOException;
import java.io.Serializable;
import java.util.Calendar;
import java.util.Map;
import java.util.Timer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Values;
import backtype.storm.utils.Utils;


public class HbaseSpout extends BaseRichSpout {
	/**
	 * 
	 */
	private static final long serialVersionUID = -6988354073044202L;

	private static final byte[][] daysCode = new byte[][]{
		new byte[]{},//这是一个占位用的
		new byte[]{(char)49},
		new byte[]{(char)50},
		new byte[]{(char)51},
		new byte[]{(char)52},
		new byte[]{(char)53},
		new byte[]{(char)54},
		new byte[]{(char)55},
		new byte[]{(char)56},
		new byte[]{(char)57},
		
		new byte[]{(char)65},
		new byte[]{(char)66},
		new byte[]{(char)67},
		new byte[]{(char)68},
		new byte[]{(char)69},
		
		new byte[]{(char)70},
		new byte[]{(char)71},
		new byte[]{(char)72},
		new byte[]{(char)73},
		new byte[]{(char)74},
		new byte[]{(char)75},
		new byte[]{(char)76},
		new byte[]{(char)77},
		new byte[]{(char)78},
		new byte[]{(char)79},
		
		new byte[]{(char)80},
		new byte[]{(char)81},
		new byte[]{(char)82},
		new byte[]{(char)83},
		new byte[]{(char)84},
		new byte[]{(char)85},
		new byte[]{(char)86},
		
	};

	public HbaseTupleProducer getTupleProducer() {
		return tupleProducer;
	}

	public void setTupleProducer(HbaseTupleProducer tupleProducer) {
		this.tupleProducer = tupleProducer;
	}

	private static final Logger LOG = LoggerFactory.getLogger(HbaseSpout.class);

	
	private boolean distributed = true;


	private SpoutOutputCollector collector;

	
	private boolean hasFailures = false;
	public final Serializable recoveryMutex = "RECOVERY_MUTEX";
	private Timer recoveryTimer = null;
	private long recoveryPeriod = -1; // default to disabled


	private HbaseTupleProducer tupleProducer;


	private HTable table;


	private Scan scan;


	private ResultScanner scanner;

	private Calendar calendar;
	

	/**
	 * <code>ISpout</code> implementation.
	 * <p/>
	 * Connects the JMS spout to the configured JMS destination
	 * topic/queue.
	 * 
	 */
	@SuppressWarnings("rawtypes")
	public void open(Map conf, TopologyContext context,
			SpoutOutputCollector collector) {
		
		if(this.tupleProducer == null){
			throw new IllegalStateException("Hbase Tuple Producer has not been set.");
		}
		Integer topologyTimeout = (Integer)conf.get("topology.message.timeout.secs");
		// TODO fine a way to get the default timeout from storm, so we're not hard-coding to 30 seconds (it could change)
		topologyTimeout = topologyTimeout == null ? 30 : topologyTimeout;
		if( (topologyTimeout.intValue() * 1000 )> this.recoveryPeriod){
		    LOG.warn("*** WARNING *** : " +
		    		"Recovery period ("+ this.recoveryPeriod + " ms.) is less then the configured " +
		    		"'topology.message.timeout.secs' of " + topologyTimeout + 
		    		" secs. This could lead to a message replay flood!");
		}
		
		Configuration hconf = new Configuration();
		hconf.set("hbase.zookeeper.quorum", "bjtest1hadoop01,bjtest1hadoop02,bjtest1hadoop03");
		try {
			table = new HTable(hconf , "object_relation_201405");
			scan = new Scan(new byte[]{(char)51}, new byte[]{(char)56});
			scan.setCaching(2000);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		calendar = Calendar.getInstance();
		this.collector = collector;
		

	}

	public void close() {
		try {
			scanner.close();
			table.close();
		} catch (IOException e) {
			LOG.error("close htable error", e);
			
		}
		
	}

	public void nextTuple() {
		calendar.setTimeInMillis(System.currentTimeMillis());;
		try {
//			scan.addFamily(daysCode[calendar.get(Calendar.DAY_OF_MONTH)]);
			scanner = table.getScanner(scan);
		} catch (IOException e2) {
			
		}
		Result next = null;
		try {
			next = scanner.next();
		} catch (IOException e1) {
			LOG.error("scanner next error", e1);
		} finally {
			scanner.close();
		}
		
		if (next == null ) {
			Utils.sleep(50);
		} else {

			LOG.debug("sending tuple: " + next);
			// get the tuple from the handler

			Values vals;
			try {
				vals = this.tupleProducer.toTuple(next);
				this.collector.emit(vals);
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}

		}

	}

	/*
	 * Will only be called if we're transactional or not AUTO_ACKNOWLEDGE
	 */
	public void ack(Object msgId) {

		

	}

	/*
	 * Will only be called if we're transactional or not AUTO_ACKNOWLEDGE
	 */
	public void fail(Object msgId) {}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		this.tupleProducer.declareOutputFields(declarer);

	}

	/**
         * Returns <code>true</code> if the spout has received failures 
         * from which it has not yet recovered.
         */
	public boolean hasFailures(){
	        return this.hasFailures;
	}
	
	protected void recovered(){
	        this.hasFailures = false;
	}
	
	/**
	 * Sets the periodicity of the timer task that 
	 * checks for failures and recovers the JMS session.
	 * 
	 * @param period
	 */
	public void setRecoveryPeriod(long period){
	    this.recoveryPeriod = period;
	}
	
	public boolean isDistributed() {
		return this.distributed;
	}
	
	/**
	 * Sets the "distributed" mode of this spout.
	 * <p/>
	 * If <code>true</code> multiple instances of this spout <i>may</i> be
	 * created across the cluster (depending on the "parallelism_hint" in the topology configuration).
	 * <p/>
	 * Setting this value to <code>false</code> essentially means this spout will run as a singleton 
	 * within the cluster ("parallelism_hint" will be ignored).
	 * <p/>
	 * In general, this should be set to <code>false</code> if the underlying JMS destination is a 
	 * topic, and <code>true</code> if it is a JMS queue.
	 * 
	 * @param distributed
	 */
	public void setDistributed(boolean distributed){
		this.distributed = distributed;
	}

	
}